iT邦幫忙

第 11 屆 iThome 鐵人賽
DAY 10
1
Software Development

Python 程式交易 30 天新手入門系列 第 10 篇

Day-10 資料保存：使用 MySQL 資料庫

11th鐵人賽 python3 程式交易 sqlalchemy pymysql
16869 瀏覽

  • 分享至 

  • xImage
    •  

安裝所需套件

使用 pip 安裝

# 資料庫
pip install pymysql sqlalchemy

下載資料庫管理軟體 HeidiSQL 並安裝
https://www.heidisql.com/download.php

安裝 MariaDB 10.4.6

使用 Docker 快速建立 MySQL 伺服器

下載 MariaDB 的 Docker 影像

https://hub.docker.com/_/mariadb

docker pull mariadb:10.4.7

建立 MariaDB 的 Docker 容器

# 建立一個帳號密碼均為 root，且有一個名為 trader 資料庫的 MySQL 伺服器
docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=trader --name mariadb mariadb:10.4.7

確認容器是否正常運行

docker ps -a

Docker 容器清單
容器狀態應有 UP 字串，代表容器正在運行。

資料庫操作

為了降低困難度，使用 SQLAlchemy ORM 來簡化操作。

建立資料表

USE test;
CREATE TABLE `stocks` (
	`id` INT(11) NOT NULL AUTO_INCREMENT,
	`code` CHAR(4) NOT NULL COLLATE 'utf8mb4_unicode_ci',
	`name` VARCHAR(32) NOT NULL COLLATE 'utf8mb4_unicode_ci',
	`created_at` TIMESTAMP NOT NULL DEFAULT current_timestamp(),
	`updated_at` TIMESTAMP NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
	`deleted_at` TIMESTAMP NULL DEFAULT NULL,
	PRIMARY KEY (`id`)
)
COLLATE='utf8mb4_unicode_ci'
ENGINE=InnoDB;

資料表格式

建立連線並寫入資料

import datetime

import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema

def main():
    username = 'root'     # 資料庫帳號
    password = 'root'     # 資料庫密碼
    host = 'localhost'    # 資料庫位址
    port = '3306'         # 資料庫埠號
    database = 'trader'   # 資料庫名稱
    # 建立連線引擎
    engine = sqlalchemy.create_engine(
        f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
    )
    # 取得資料庫元資料
    metadata = sqlalchemy.schema.MetaData(engine)
    # 產生自動對應參照
    automap = sqlalchemy.ext.automap.automap_base()
    automap.prepare(engine, reflect=True)
    # 準備 ORM 連線
    session = sqlalchemy.orm.Session(engine)

    # 載入 stocks 資料表資訊
    sqlalchemy.Table('stocks', metadata, autoload=True)
    # 取出對應 stocks 資料表的類別
    Stock = automap.classes['stocks']

    try:
        # 執行原生 SQL 命令清空資料庫
        session.execute('TRUNCATE TABLE stocks')
        # 送出執行命令
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('清空資料失敗')

    # 進入交易模式
    try:
        # 建立第一筆資料
        stock = Stock()
        stock.code = '1101'
        stock.name = '台泥'
        session.add(stock)

        # 建立第二筆資料
        stock = Stock()
        stock.code = '1102'
        stock.name = '亞泥'
        session.add(stock)

        # 建立第三筆資料
        stock = Stock()
        stock.code = '1103'
        stock.name = '嘉泥'
        session.add(stock)

        # 建立第四筆資料
        stock = Stock()
        stock.code = '1201'
        stock.name = '味全'
        session.add(stock)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('新增資料失敗')
        loguru.logger.error(e)

    # 關閉連線
    session.close()

if __name__ == '__main__':
    loguru.logger.add(
        f'{datetime.date.today():%Y%m%d}.log',
        rotation='1 day',
        retention='7 days',
        level='DEBUG'
    )
    main()

資料表內容

讀取資料

import datetime

import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema

def main():
    username = 'root'     # 資料庫帳號
    password = 'root'     # 資料庫密碼
    host = 'localhost'    # 資料庫位址
    port = '3306'         # 資料庫埠號
    database = 'trader'   # 資料庫名稱
    # 建立連線引擎
    engine = sqlalchemy.create_engine(
        f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
    )
    metadata = sqlalchemy.schema.MetaData(engine)
    automap = sqlalchemy.ext.automap.automap_base()
    automap.prepare(engine, reflect=True)
    session = sqlalchemy.orm.Session(engine)

    sqlalchemy.Table('stocks', metadata, autoload=True)
    Stock = automap.classes['stocks']

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    loguru.logger.info('使用過濾條件取出資料表資料')
    results = session.query(Stock).filter(
        Stock.code == '1102'
    ).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    loguru.logger.info('使用多重過濾條件 OR 取出資料表資料')
    results = session.query(Stock).filter(
        sqlalchemy.or_(
            Stock.code.like('%02'),
            Stock.code.like('12%')
        )
    ).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    loguru.logger.info('使用多重過濾條件 AND 取出資料表資料')
    results = session.query(Stock).filter(
        sqlalchemy.and_(
            Stock.code.like('%2'),
            Stock.name.like('%泥')
        )
    ).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    loguru.logger.info('使用排序取出資料表資料')
    results = session.query(Stock).order_by(Stock.code.desc()).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    session.close()

if __name__ == '__main__':
    loguru.logger.add(
        f'{datetime.date.today():%Y%m%d}.log',
        rotation='1 day',
        retention='7 days',
        level='DEBUG'
    )
    main()

更新資料

import datetime

import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema

def main():
    username = 'root'     # 資料庫帳號
    password = 'root'     # 資料庫密碼
    host = 'localhost'    # 資料庫位址
    port = '3306'         # 資料庫埠號
    database = 'trader'   # 資料庫名稱
    # 建立連線引擎
    engine = sqlalchemy.create_engine(
        f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
    )
    metadata = sqlalchemy.schema.MetaData(engine)
    automap = sqlalchemy.ext.automap.automap_base()
    automap.prepare(engine, reflect=True)
    session = sqlalchemy.orm.Session(engine)

    sqlalchemy.Table('stocks', metadata, autoload=True)
    Stock = automap.classes['stocks']

    loguru.logger.info('----- 更新單筆資料 -----')
    try:
        stock = session.query(Stock).filter(
            Stock.code == '1102'
        ).one()
        stock.name = stock.name + '（水泥產業）'
        session.add(stock)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('更新資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    loguru.logger.info('----- 更新多筆資料 -----')
    try:
        session.query(Stock).filter(
            sqlalchemy.and_(
                Stock.code.like('11%'),
                Stock.code != '1102'
            )
        ).update({
            Stock.name: Stock.name + '（水泥產業）'
        }, synchronize_session=False)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('更新資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    session.close()

if __name__ == '__main__':
    loguru.logger.add(
        f'{datetime.date.today():%Y%m%d}.log',
        rotation='1 day',
        retention='7 days',
        level='DEBUG'
    )
    main()

刪除資料

import datetime

import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema

def main():
    username = 'root'     # 資料庫帳號
    password = 'root'     # 資料庫密碼
    host = 'localhost'    # 資料庫位址
    port = '3306'         # 資料庫埠號
    database = 'trader'   # 資料庫名稱
    # 建立連線引擎
    engine = sqlalchemy.create_engine(
        f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
    )
    metadata = sqlalchemy.schema.MetaData(engine)
    automap = sqlalchemy.ext.automap.automap_base()
    automap.prepare(engine, reflect=True)
    session = sqlalchemy.orm.Session(engine)

    sqlalchemy.Table('stocks', metadata, autoload=True)
    Stock = automap.classes['stocks']
    
    loguru.logger.info('----- 刪除單筆資料 -----')
    try:
        stock = session.query(Stock).filter(
            Stock.code == '1102'
        ).one()
        session.delete(stock)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('刪除資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')
    
    loguru.logger.info('----- 刪除多筆資料 -----')
    try:
        session.query(Stock).filter(
            Stock.code.like('11%')
        ).delete(synchronize_session=False)

        # 寫入資料庫
        session.commit()
    except Exception as e:
        # 發生例外錯誤，還原交易
        session.rollback()
        loguru.logger.error('刪除資料失敗')
        loguru.logger.error(e)

    loguru.logger.info('取出資料表所有資料')
    results = session.query(Stock).all()
    for stock in results:
        loguru.logger.info(f'{stock.code} {stock.name}')

    session.close()

if __name__ == '__main__':
    loguru.logger.add(
        f'{datetime.date.today():%Y%m%d}.log',
        rotation='1 day',
        retention='7 days',
        level='DEBUG'
    )
    main()

Would You Like To Know More?

https://pymysql.readthedocs.io/
https://docs.sqlalchemy.org/

團隊系列文：

CSScoke - 金魚都能懂的這個網頁畫面怎麼切 - 金魚都能懂了你還怕學不會嗎
Clarence - LINE bot 好好玩 30 天玩轉 LINE API
Hina Hina - 陣列大亂鬥
King Tzeng - IoT沒那麼難！新手用JavaScript入門做自己的玩具
Vita Ora - 好 Js 不學嗎 !? JavaScript 入門中的入門。
TaTaMo - 用Python開發的網頁不能放到Github上？Lektor說可以！！


上一篇
Day-09 資料蒐集：取得個股當日基本資訊 + 解析 PDF
下一篇
Day-11 資料保存：取得 2019 年每日加權指數盤後資訊並繪製線圖
系列文
Python 程式交易 30 天新手入門 共 30 篇
  1. 26
    Day-26 回測系統：時序運行模組
  2. 27
    Day-27 回測系統：策略運行模組
  3. 28
    Day-28 回測系統：交易執行模組
  4. 29
    Day-29 回測系統：交易統計模組
  5. 30
    Day-30 發大財？
完整目錄
圖片
  直播研討會
圖片
{{ item.channelVendor }} {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言
iThome鐵人賽
參賽組數
1064 組
團體組數
40 組
累計文章數
22200 篇
完賽人數
602 人
iT+看影片追技術 看影片追技術 看更多
熱門tag 看更多
15th鐵人賽 16th鐵人賽 13th鐵人賽 14th鐵人賽 12th鐵人賽 11th鐵人賽 鐵人賽 2019鐵人賽 javascript 2018鐵人賽 python 2017鐵人賽 windows php c# windows server linux css react vue.js
熱門問題
熱門回答
熱門文章
IT邦幫忙